package com.thomsonreuter.search.compare.excutor;

import com.fasterxml.jackson.core.*;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Multimap;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.jayway.jsonpath.JsonPath;
import okhttp3.*;


import java.io.*;
import java.net.SocketException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;

import java.util.concurrent.TimeUnit;


public class ElasticExecutor {


    private static OkHttpClient client =
            new OkHttpClient.Builder()
                    .connectTimeout(0, TimeUnit.SECONDS)//设置连接超时时间
                    .readTimeout(0, TimeUnit.SECONDS)//设置读取超时时间
                    .build();


    public static Map<String, Integer> getSampleSummary(String content) {
        List<Map<String, Object>> buckets = JsonPath.read(content, "$..buckets.*");
        return buckets.stream()
                .collect(ImmutableMap.toImmutableMap(m -> (String) m.get("key"), m -> (Integer) m.get("doc_count")));
    }

    public static List<String> getDocumentTypes(String content) {
        return JsonPath.read(content, "$.._type");
    }

    public static List<String> getDocumentIds(String content) {
        return JsonPath.read(content, "$.._id");

    }


    public static String retriveQueryResponseContent(final String index) throws IOException {
        String queryStr = QueryTemplate.getQueryStr();
        final Request request = new Request.Builder().url(index + "/_search").post(RequestBody.create(MediaType.get("text/plain"), queryStr)).build();
        try (Response response = client.newCall(request).execute()) {

            return response.code() == 200 ? response.body().string() : null;
        }
    }


    /**
     * 根据documentId获取--source的内容,供后续json串的比较
     *
     * @param index
     * @param documentId
     * @return
     * @throws IOException
     */
    public static String getDocContentById(final String index, final String type, final String documentId) throws IOException {
        final Request request = new Request.Builder().url(index + "/" + type + "/" + documentId).get().build();
        try (Response response = client.newCall(request).execute()) {
            if (response.code() != 200) {
                System.out.println("Search document " + documentId + " failed. Response status is:" + response.code());
                return null;
            }
            String content = response.body().string();
            Gson gson = new GsonBuilder().disableHtmlEscaping().create();

            final Map map = gson.fromJson(content, Map.class);

            final Object source = map.get("_source");

            return source != null ? gson.toJson(source) : null;
        }
    }


    public static String getFirstPageWithScroll(final String index) throws IOException {
        String queryStr = QueryTemplate.getPageQueryStr();
        StringBuilder scrollindex = new StringBuilder();
        scrollindex.append(index);
        scrollindex.append("/_search");
        scrollindex.append("?scroll=");
        scrollindex.append("5m");

        final Request request = new Request.Builder().url(scrollindex.toString()).post(RequestBody.create(MediaType.get("text/plain"), queryStr)).build();
        try (Response response = client.newCall(request).execute()) {

            return response.code() == 200 ? response.body().string() : null;
        }
    }

    public static String getNextPageWithScrollId(final String index, final String scrollId) throws IOException {
        String queryStr = QueryTemplate.getNextPageQueryStr(scrollId);
        StringBuilder scrollindex = new StringBuilder();
        scrollindex.append(index);
        scrollindex.append("/_search");
        scrollindex.append("?scroll=");
        scrollindex.append("5m");

        final Request request = new Request.Builder().url(scrollindex.toString()).post(RequestBody.create(MediaType.get("text/plain"), queryStr)).build();
        try (Response response = client.newCall(request).execute()) {

            return response.code() == 200 ? response.body().string() : null;
        }
    }

    public static String getDuplicateISINJson(final String index) throws IOException {
        String queryStr = QueryTemplate.getPageQueryStr();
        StringBuilder scrollindex = new StringBuilder();
        scrollindex.append(index);
        scrollindex.append("/_search");


        final Request request = new Request.Builder().url(scrollindex.toString()).post(RequestBody.create(MediaType.get("text/plain"), queryStr)).build();
        try (Response response = client.newCall(request).execute()) {
            InputStream is = response.body().byteStream();
            return response.code() == 200 ? response.body().string() : null;
        }
    }

    public static Response getDuplicateISINJsonStream(final String index) throws IOException {
        String queryStr = QueryTemplate.getPageQueryStr();
        StringBuilder scrollindex = new StringBuilder();

        scrollindex.append(index);
        scrollindex.append("/_search");



        Response response = null;
        while (true){
            try {
                final Request request = new Request.Builder().url(scrollindex.toString()).post(RequestBody.create(MediaType.get("text/plain"), queryStr)).build();
                response =   client.newCall(request).execute();
                break;
            } catch (SocketException se) {

            }
        }
        return response;


    }



    public static String getCategoryPath(String RootDir, String prefix, String category){
        StringBuilder fullpath = new StringBuilder();
        fullpath.append(RootDir );
        fullpath.append(prefix);
        fullpath.append(category);
        fullpath.append(".json");
        return fullpath.toString();
    }

    public static void analysisDuplicatISIN(InputStream is,
                                            String rootDir,
                                            String prefix,
                                            String fullcatIndex,
                                            String fulleikonIndex) throws IOException {
        JsonFactory factory = new JsonFactory();
        JsonParser jsonParser = factory.createParser(is);
        Map<String, JsonGenerator> jsonGeneratorMap = new HashMap<>();
        Map<String, String> category2PathMap = new HashMap<>();


        Multimap<String, Multimap<String, Map<String, String>>> category2Isins
                = ArrayListMultimap.create();
        Map<String, Long> summary = new LinkedHashMap<>();

        long isinTotalNum = 0;
        while (!jsonParser.isClosed()) {
            JsonToken jsonToken = jsonParser.nextToken();
            if(jsonToken == null){
                break;
            }

            if (JsonToken.FIELD_NAME.equals(jsonToken)) {
                String fieldName = jsonParser.getCurrentName();

                if ("key".equals(fieldName)) {
                    jsonParser.nextToken();

                    String ISIN = jsonParser.getValueAsString();
                    analysisOneISIN(ISIN,
                            fullcatIndex,
                            fulleikonIndex,
                            category2Isins,
                            summary);

                    isinTotalNum++;

                    if( isinTotalNum % 1000 == 0) {
                        for (Map.Entry<String, Collection<Multimap<String, Map<String, String>>>> entry : category2Isins.asMap().entrySet()) {
                            String category = entry.getKey();


                            JsonGenerator jsonGenerator = jsonGeneratorMap.computeIfAbsent(category, k ->
                            {
                                JsonGenerator jsonGeneratorInside = null;
                                try {
                                    String fileName =  getCategoryPath(rootDir,prefix,category);
                                    jsonGeneratorInside = factory.createGenerator(new File(fileName),
                                                                                  JsonEncoding.UTF8);
                                    jsonGeneratorInside = jsonGeneratorInside.useDefaultPrettyPrinter();
                                    jsonGeneratorInside.writeStartArray();
                                    category2PathMap.put(category,fileName);

                                } catch (IOException ioe) {
                                    ioe.printStackTrace();
                                }

                                return jsonGeneratorInside;
                            });

                            Collection<Multimap<String, Map<String, String>>> isin2docsMap = entry.getValue();

                            dumpCurrentDocs(factory, jsonGenerator, summary, category, isin2docsMap);

                        }
                        category2Isins.clear();
                    }
                }
            }
        }

        for (Map.Entry<String, Collection<Multimap<String, Map<String, String>>>> entry : category2Isins.asMap().entrySet()) {
            String category = entry.getKey();

            Collection<Multimap<String, Map<String, String>>> isin2docsMap = entry.getValue();

            JsonGenerator jsonGenerator = jsonGeneratorMap.computeIfAbsent(category, k ->
            {
                JsonGenerator jsonGeneratorInside = null;
                try {
                    String fileName =  getCategoryPath(rootDir,prefix,category);
                    jsonGeneratorInside = factory.createGenerator(new File(fileName),
                            JsonEncoding.UTF8);
                    jsonGeneratorInside = jsonGeneratorInside.useDefaultPrettyPrinter();
                    category2PathMap.put(category,fileName);

                } catch (IOException ioe) {
                    ioe.printStackTrace();
                }

                return jsonGeneratorInside;
            });


            dumpCurrentDocs(factory, jsonGenerator, summary, category, isin2docsMap);

        }

        jsonGeneratorMap.forEach((key,value) -> {
            try {
                value.writeEndArray();
                value.close();
            }catch(IOException ioe ){
                ioe.printStackTrace();
            }
        });

        jsonParser.close();

        //generate the final file:
        StringBuilder fullpath = new StringBuilder();
        fullpath.append(rootDir );
        fullpath.append(prefix);
        fullpath.append("final");
        fullpath.append(".json");
        JsonGenerator jsonGenerator = factory.createGenerator(new File(fullpath.toString()),
                JsonEncoding.UTF8);

        jsonGenerator = jsonGenerator.useDefaultPrettyPrinter();
        JsonGenerator jsonGeneratorFinal = jsonGenerator;
        jsonGeneratorFinal.writeStartObject();
        jsonGeneratorFinal.writeStringField("totalNumber",String.valueOf(isinTotalNum));
        //summary.put("totalNumber", isinTotalNum);

        summary.forEach((key,value) ->{
            System.out.println(key + ":" + value);
            try {
                jsonGeneratorFinal.writeStringField(key, String.valueOf(value));
            }catch(IOException ioe){
                ioe.printStackTrace();
            }
        });
        jsonGeneratorFinal.writeFieldName("categorydetail");

        jsonGeneratorFinal.writeStartObject();
        category2PathMap.forEach((key,value) ->{
            try{
                jsonGeneratorFinal.writeFieldName(key);
                File file = new File(value);
                JsonParser jsonParserInternal = factory.createParser(file);
                jsonMerge(jsonGeneratorFinal, jsonParserInternal);
                jsonParserInternal.close();

            }catch (IOException ioe){
                ioe.printStackTrace();
            }
        });
        jsonGeneratorFinal.writeEndObject();
        jsonGeneratorFinal.writeEndObject();
        jsonGeneratorFinal.close();


    }

    public static void jsonMerge(JsonGenerator jsonGeneratorFinal, JsonParser jsonParserInternal) throws IOException {
        try {
            JsonToken jsonToken = null;
            while (!jsonParserInternal.isClosed()) {
                jsonToken = jsonParserInternal.nextToken();
                if (jsonToken == null)
                    break;
                if (jsonToken.equals(JsonToken.START_ARRAY)) {
                    jsonGeneratorFinal.writeStartArray();
                } else if (jsonToken.equals(JsonToken.START_OBJECT)) {
                    jsonGeneratorFinal.writeStartObject();
                } else if (jsonToken.equals(JsonToken.FIELD_NAME)) {
                    String fieldName = jsonParserInternal.getCurrentName();
                    jsonGeneratorFinal.writeFieldName(fieldName);
                } else if (jsonToken.equals(JsonToken.END_ARRAY)) {
                    jsonGeneratorFinal.writeEndArray();
                } else if (jsonToken.equals(JsonToken.END_OBJECT)) {
                    jsonGeneratorFinal.writeEndObject();
                } else if (jsonToken.equals(JsonToken.VALUE_STRING)) {
                    String value = jsonParserInternal.getValueAsString();
                    jsonGeneratorFinal.writeString(value);
                }else if(jsonToken.equals(JsonToken.VALUE_NULL)){
                    jsonGeneratorFinal.writeNull();
                }

            }
        }
        catch(IOException ioe){
            ioe.printStackTrace();
        }
    }

    private static void dumpCurrentDocs(JsonFactory factory,
                                        JsonGenerator jsonGenerator,
                                        Map<String, Long> summary,
                                        String category,
                                        Collection<Multimap<String, Map<String, String>>> isin2docsMap) throws IOException {
        Long currentNumber = summary.get(category);
        if(isin2docsMap != null ) {
            int nowsize = isin2docsMap.size();
            if(currentNumber == null)
            {
                currentNumber = new Long(0);
            }
            currentNumber += nowsize;
            summary.put(category, currentNumber);

            for (Multimap<String, Map<String, String>> isin2docs : isin2docsMap) {
                for (Map.Entry<String, Collection<Map<String, String>>> isin2docsEntry : isin2docs.asMap().entrySet()) {
                    System.out.println("        ISIN is " + isin2docsEntry.getKey());
                    writeOneEntry(jsonGenerator, isin2docsEntry);

                }
            }

        }
    }

    public static void analysisOneISIN(String ISIN,
                                       String catindex,
                                       String eikonIndex,
                                       Multimap<String, Multimap<String, Map<String, String>>> category2Isins,
                                       Map<String, Long> summary) throws IOException {


        //get documents which contains this isin
        String docIsin = ElasticExecutor.getDocumentsWithISINJson(catindex, ISIN);

        List<String> sources = JsonPath.read(docIsin, "$..Source");
        List<String> _ids = JsonPath.read(docIsin, "$.._id");
        List<String> PermIds = JsonPath.read(docIsin, "$..PermID");
        List<String> BusinessEntities = JsonPath.read(docIsin, "$..BusinessEntity");
        List<Long> timestamps = JsonPath.read(docIsin, "$.._timestamp");
        if (sources.size() <= 1) {
            return;
        }

        Multimap<String, Integer> type2indexmap = ArrayListMultimap.create();

        int i = 0;
        for (String source : sources) {
            i++;
            if (source.equalsIgnoreCase("CCC")) {
                type2indexmap.put("CCC", Integer.valueOf(i));
            } else {
                type2indexmap.put("CCCCL3", Integer.valueOf(i));
            }
        }

        Multimap<String, Map<String, String>> isin2docs = ArrayListMultimap.create();
        int permSize = PermIds.size();
        int businessEntitySize = BusinessEntities.size();
        for (int idx = 0; idx < _ids.size(); idx++) {
            Map<String, String> docMap = new HashMap<>();
            docMap.put("id", _ids.get(idx));
            if (idx < permSize) {
                docMap.put("PermID", PermIds.get(idx));
            }

            if (idx < businessEntitySize) {
                docMap.put("BusinessEntity", BusinessEntities.get(idx));
            }


            docMap.put("Source", sources.get(idx));
            DateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:sss");

            Long epoch = timestamps.get(idx);
            long epochl = epoch.longValue();
            Date d = new Date(epochl);
            String t = df.format(d);
            docMap.put("timestamp", t);


            isin2docs.put(ISIN, docMap);

        }

        if (type2indexmap.asMap().size() > 1) {
            category2Isins.put("CCCCL3Duplication", isin2docs);
        } else if (type2indexmap.asMap().size() == 1) {

            if (sources.get(0).equalsIgnoreCase("CCC")) {
                //it is a ccc document duplication:
                category2Isins.put("CCCDuplication", isin2docs);
            } else {


                boolean isAllDocInEikon = true;
                boolean isIsinDuplicateInEikon = true;
                for (Map.Entry<String, Collection<Map<String, String>>> isin2docsEntry : isin2docs.asMap().entrySet()) {
                    for (Map<String, String> doc : isin2docsEntry.getValue()) {
                        String eikonDoc = ElasticExecutor.getDocumentsWithIdJson(eikonIndex, doc.get("id"));

                        List<String> eikonDocIds = JsonPath.read(eikonDoc, "$.._id");
                        List<String> eikonISIN = JsonPath.read(eikonDoc, "$..ISIN");
                        if (eikonDocIds.size() < 1) {
                            isAllDocInEikon = false;
                        }
                        if (eikonISIN.size() == 1) {
                            if (!eikonISIN.get(0).equalsIgnoreCase(ISIN)) {
                                isIsinDuplicateInEikon = false;
                            }
                        } else {
                            isIsinDuplicateInEikon = false;
                        }
                    }
                }
                if (isAllDocInEikon && isIsinDuplicateInEikon) {
                    category2Isins.put("CL3DuplicationEikonDuplication", isin2docs);
                } else {
                    category2Isins.put("CL3Duplication", isin2docs);
                }

            }
        }
    }

    public static void writeOneEntry(JsonGenerator jsonGenerator,
                                     Map.Entry<String, Collection<Map<String, String>>> isin2docsEntry) throws IOException {
        jsonGenerator.writeStartObject();
        jsonGenerator.writeFieldName(isin2docsEntry.getKey());
        jsonGenerator.writeStartArray();
        for (Map<String, String> doc : isin2docsEntry.getValue()) {

            jsonGenerator.writeStartObject();
            jsonGenerator.writeStringField("id", doc.get("id"));
            jsonGenerator.writeStringField("PermID", doc.get("PermID"));
            jsonGenerator.writeStringField("BusinessEntity", doc.get("BusinessEntity"));
            jsonGenerator.writeStringField("Source", doc.get("Source"));
            jsonGenerator.writeStringField("timestamp", doc.get("timestamp"));
            jsonGenerator.writeEndObject();
        }
        jsonGenerator.writeEndArray();
        jsonGenerator.writeEndObject();
    }

    public static void saveInputStream2Local(InputStream is, String filepath) throws IOException
    {
        FileOutputStream output = new FileOutputStream(filepath);
        int bufferSize = 1024;
        byte[] buffer = new byte[bufferSize];
        int len;
        while ((len = is.read(buffer)) != -1)
        {
            output.write(buffer, 0, len);
        }
        output.close();
    }

    public static List<String> getDuplicateISINs(String content) {
        return JsonPath.read(content, "$..key");

    }

    public static String getDocumentsWithISINJson(final String index, final String isin) throws IOException {
        String queryStr = QueryTemplate.getIsinDocQueryStr(isin);
        StringBuilder scrollindex = new StringBuilder();
        scrollindex.append(index);
        scrollindex.append("/_search");


        final Request request = new Request.Builder().url(scrollindex.toString()).post(RequestBody.create(MediaType.get("text/plain"), queryStr)).build();
        try (Response response = client.newCall(request).execute()) {
            return response.code() == 200 ? response.body().string() : null;
        }
    }

    public static String getDocumentsWithIdJson(final String index, final String docid) throws IOException {
        String queryStr = QueryTemplate.getEikonDocQueryStr(docid);
        StringBuilder scrollindex = new StringBuilder();
        scrollindex.append(index);
        scrollindex.append("/_search");


        final Request request = new Request.Builder().url(scrollindex.toString()).post(RequestBody.create(MediaType.get("text/plain"), queryStr)).build();
        try (Response response = client.newCall(request).execute()) {
            return response.code() == 200 ? response.body().string() : null;
        }
    }

}
